package com.example.network.demos.tcp.client.nanchu;

import cn.hutool.core.thread.ThreadUtil;
import com.alibaba.fastjson2.JSON;
import com.example.network.demos.config.NetworkConfig;
import com.example.network.demos.tcp.domain.TCPMessage;
import com.example.network.demos.tcp.variable.Variable;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Netty TCP客户端
 *
 * @author chenlei
 */
@Slf4j
@Component
public class NettyTcpClientNanchu implements CommandLineRunner {

    // ✅ 新增：保存 Channel 的引用
    private static volatile Channel channel;
    final AtomicInteger connectTimes = new AtomicInteger(0);

    @Override
    public void run(String... args) {
        ThreadUtil.execAsync(new Runnable() {
            @Override
            public void run() {

                EventLoopGroup group = new NioEventLoopGroup();
                try {
                    Bootstrap bootstrap = new Bootstrap();
                    bootstrap.group(group)
                            .channel(NioSocketChannel.class)
                            .option(ChannelOption.TCP_NODELAY, true)
                            .handler(new ChannelInitializer<SocketChannel>() {
                                @Override
                                protected void initChannel(SocketChannel ch) {
                                    ch.pipeline().addLast(new NettyTcpClientHandlerNanchu(bootstrap, NetworkConfig.getTcpClientIPOther(), Integer.parseInt(NetworkConfig.getTcpClientPortNanchu()), NettyTcpClientNanchu.this));
                                }
                            });

                    final AtomicInteger connectTimes = new AtomicInteger(0);
                    bootstrap.connect(NetworkConfig.getTcpClientIPOther(), Integer.parseInt(NetworkConfig.getTcpClientPortNanchu())).addListener((ChannelFutureListener) future -> {
                        if (future.isSuccess()) {
                            log.info("成功连接到端口: {}", Integer.parseInt(NetworkConfig.getTcpClientPortNanchu()));
                            // ✅ 连接成功时保存 channel
                            channel = future.channel();
                        } else {
                            log.error("连接到服务器时出错，端口: {}", Integer.parseInt(NetworkConfig.getTcpClientPortNanchu()));
                            connectTimes.incrementAndGet();
//                    future.channel().eventLoop().schedule(() -> bootstrap.connect(host, port), 10, TimeUnit.SECONDS);
                            reconnect(bootstrap, NetworkConfig.getTcpClientIPOther(), Integer.parseInt(NetworkConfig.getTcpClientPortNanchu()));
                        }
                    });
                } catch (Exception e) {
                    log.error("Netty 客户端出错", e);
                }
            }
        });

    }

    // ✅ 新增：发送消息的方法
    public boolean sendMessage(TCPMessage tcpMessage) {
        String message = JSON.toJSONString(tcpMessage) + "\n";
        if (isConnected()) {
            channel.writeAndFlush(Unpooled.copiedBuffer(message.getBytes())).addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    log.info("消息发送成功: {}", message);
                } else {
                    log.error("消息发送失败: {}", message, future.cause());
                }
            });
            return true;
        } else {
            log.warn("通道未激活，无法发送消息");
            return false;
        }
    }

    // ✅ 新增：发送消息的方法`
    public TCPMessage getMessage(TCPMessage tcpMessage) {
        String id = tcpMessage.getId();
        String message = JSON.toJSONString(tcpMessage) + "\n";
        if (isConnected()) {
            channel.writeAndFlush(Unpooled.copiedBuffer(message.getBytes())).addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    log.info("消息发送成功: {}", message);
                } else {
                    log.error("消息发送失败: {}", message, future.cause());
                }
            });
            TCPMessage getMessage = new TCPMessage();
            long startTime = System.currentTimeMillis();
            while (true) {
                getMessage = Variable.tcpMessageConcurrentHashMap.get(id);

                if (getMessage != null) {
                    return getMessage; // 获取到消息，立即返回
                }

                // 判断是否超时
                if (System.currentTimeMillis() - startTime >= NetworkConfig.getTcpTimeOut()) {
                    log.error("物联感知网络连接超时");
                    return null;
                }
            }
        } else {
            log.warn("通道未激活，无法发送消息");
            return null;
        }
    }

    // ✅ 可选：提供获取连接状态的方法
    public boolean isConnected() {
        return channel != null && channel.isActive();
    }

    // ✅ 可选：关闭连接的方法
    public void shutdown() {
        if (channel != null && channel.isActive()) {
            channel.close();
        }
    }

    public void reconnect(Bootstrap bootstrap, String host, int port) {
        log.info("正在进行第 {} 次重连...", connectTimes.incrementAndGet());
        try {
            bootstrap.connect(host, port).addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    log.info("重连成功");
                    connectTimes.set(0); // 重置计数器
                    channel = future.channel();   // 更新 channel
                } else {
                    log.warn("重连失败，将在30秒后再次尝试");
                    future.channel().eventLoop()
                            .schedule(() -> reconnect(bootstrap, host, port), 30, TimeUnit.SECONDS);
                }
            });
        } catch (Exception e) {
            log.error("重连时发生异常，端口: {}", port);
            reconnect(bootstrap, host, port);
        }
    }
}
